<?php
/**
 * Copyright (c) [2019] [selibra]
 * [吴跃忠] is licensed under the Mulan PSL v1.
 * You can use this software according to the terms and conditions of the Mulan PSL v1.
 * You may obtain a copy of Mulan PSL v1 at:
 * http://license.coscl.org.cn/MulanPSL
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR
 * PURPOSE.
 * See the Mulan PSL v1 for more details.
 */

namespace Selibra\MQ\Annotations;


use AMQPChannel;
use AMQPConnection;
use AMQPEnvelope;
use AMQPExchange;
use AMQPQueue;
use Co\System;
use Selibra\Config\Config;
use Selibra\Di\AnnotationConfigure;
use Selibra\Model\Build\ConnectBuilder;
use Selibra\Tools\Console;
use Selibra\Di\AnnotationExecEntity;
use Selibra\Di\Annotations\Protocol\SelibraAnnotationInterface;
use Selibra\Di\DI;
use Swoole\Coroutine;
use Swoole\Process;
use Swoole\Timer;
use \Attribute;

/**
 * Class MQConsumer
 * @package Selibra\MQ\Annotations
 */
#[Attribute(Attribute::TARGET_METHOD)]
final class MQConsumer implements SelibraAnnotationInterface
{

    /**
     * @var string
     */
    protected $tags;

    public function __construct($tags)
    {
        $this->tags = $tags;
    }


    /**
     * @inheritDoc
     * @throws
     */
    public function exec(AnnotationExecEntity &$annotationExecEntity)
    {
        $process = new Process(function () use ($annotationExecEntity) {
            go(function () use ($annotationExecEntity) {
                $config = Config::get('rabbitmq.connect');
                $arr = [
                    'host' => $config['host'],
                    'port' => $config['port'],
                    'user' => $config['user'],
                    'password' => $config['password'],
                    'vhost' => $config['vhost'],
                ];
                //构造函数   AMQPConnection
                $con = new AMQPConnection($arr);
                try{
                    if (!$con->connect()) {
                        var_dump('连接失败1');
                    }
                }catch (\AMQPConnectionException $exception){
                    System::sleep(20);
                    if (!$con->connect()) {
                        var_dump('连接失败1');
                    }
                }


                //先通道声明--传入连接的套接字--构造函数 通过通道连接创建消息通道
                $channel = new AMQPChannel($con);

                //交换机声明--传入声明的通道-- 构造函数 通过通道连接交换机
                $exchange = new AMQPExchange($channel);

                //设置交换机名
                $exchange->setName($this->tags);//设置通道名称

                //设置连接方式--直连 [直连，主题，广播]
                $exchange->setType(AMQP_EX_TYPE_DIRECT);

                //消息持久化
                $exchange->setFlags(AMQP_DURABLE);

                //声明
                $exchange->declareExchange();

                //声明队列，绑定交换机和路由
                $queue = new AMQPQueue($channel);

                //设置队列名字
                $queue->setName($this->tags);

                //消息持久化
                $queue->setFlags(AMQP_DURABLE);

                $queue->setArgument('x-max-priority',255);

                //声明
                $queue->declareQueue();

                //绑定获取数据 参数一：交换机名  参数二：路由
                $queue->bind($this->tags, $this->tags);

                //消费，没有数据时，阻塞监听获取数据
                Console::log("Consumer", $this->tags);
                $queue->consume(function (AMQPEnvelope $event, AMQPQueue $queue) use ($annotationExecEntity) {
                    // 验证数据库是否可用 重连数据库
//                    ConnectBuilder::verifyAvailabilityRestart();

                    $body = $event->getBody();
                    $event->getConsumerTag();
                    $class = DI::getObjectContext($annotationExecEntity->getClass());
                    $method = $annotationExecEntity->getMethod()->getName();
                    $result = $class->$method($body);
                    if ($result !== false) {
                        $state = $queue->ack($event->getDeliveryTag());
                    }
                });
            });

        });
        $process->start();
    }


    public function configure(AnnotationConfigure $annotationConfigure)
    {
        // TODO: Implement configure() method.
    }
}
